package com.winit.schedule.core.localstore.leveldb;

import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;

import org.fusesource.leveldbjni.JniDBFactory;
import org.iq80.leveldb.DB;
import org.iq80.leveldb.DBIterator;
import org.iq80.leveldb.Options;
import org.iq80.leveldb.ReadOptions;
import org.iq80.leveldb.Snapshot;
import org.iq80.leveldb.WriteBatch;

import com.winit.schedule.core.exception.LocalStoreException;
import com.winit.schedule.core.localstore.AbstractLocalStore;
import com.winit.schedule.core.localstore.KVPair;
import com.winit.schedule.core.utils.Constants;
import com.winit.schedule.core.utils.FileUtils;
import com.winit.schedule.core.utils.SerializableUtil;

public class LeveldbLocalStore extends AbstractLocalStore {

    private DB                 db;

    private Options            options;
    // 保证同时只有一个线程修改
    private ReentrantLock      lock = new ReentrantLock();

    public static final String name = "leveldb";

    public LeveldbLocalStore(String storePath, String identity){
        this(new File(storePath.concat(name).concat(Constants.FILE_SEPARATOR).concat(identity)));
        getLock(dbPath.getPath());
    }

    public LeveldbLocalStore(File dbPath){
        super(dbPath);
    }

    @Override
    protected void init() {
        options = new Options();
        options.createIfMissing(true);
        options.cacheSize(10 * 1024 * 1024); // 10M
    }

    protected String getName() {
        return name;
    }

    @Override
    public void open() throws LocalStoreException {
        try {
            lock.lock();
            JniDBFactory.factory.repair(dbPath, options);
            db = JniDBFactory.factory.open(dbPath, options);
        } catch (IOException e) {
            throw new LocalStoreException(e);
        }
    }

    @Override
    public void put(String key, Object value) throws LocalStoreException {
        try {
            byte[] valueBytes = SerializableUtil.encode(value);
            db.put(key.getBytes("UTF-8"), valueBytes);
        } catch (Exception e) {
            throw new LocalStoreException(e);
        }
    }

    @Override
    public void delete(String key) throws LocalStoreException {
        try {
            if (key == null) {
                return;
            }
            db.delete(key.getBytes("UTF-8"));
        } catch (Exception e) {
            throw new LocalStoreException(e);
        }
    }

    @Override
    public void delete(List<String> keys) throws LocalStoreException {
        if (keys == null || keys.size() == 0) {
            return;
        }
        WriteBatch batch = db.createWriteBatch();
        try {

            for (String key : keys) {
                batch.delete(key.getBytes("UTF-8"));
            }
            db.write(batch);
        } catch (UnsupportedEncodingException e) {
            throw new LocalStoreException(e);
        } finally {
            try {
                batch.close();
            } catch (IOException e) {
                throw new LocalStoreException(e);
            }
        }
    }

    @Override
    public <T> List<KVPair<String, T>> fetchTop(int size, Class<T> type) throws LocalStoreException {
        Snapshot snapshot = db.getSnapshot();
        DBIterator iterator = null;
        try {
            List<KVPair<String, T>> list = new ArrayList<KVPair<String, T>>(size);
            ReadOptions options = new ReadOptions();
            options.snapshot(snapshot);
            iterator = db.iterator(options);
            for (iterator.seekToFirst(); iterator.hasNext(); iterator.next()) {
                Map.Entry<byte[], byte[]> entry = iterator.peekNext();
                String key = new String(entry.getKey(), "UTF-8");
                T value = SerializableUtil.decode(entry.getValue(), type);
                KVPair<String, T> pair = new KVPair<String, T>(key, value);
                list.add(pair);
                if (list.size() >= size) {
                    break;
                }
            }
            return list;
        } catch (Exception e) {
            throw new LocalStoreException(e);
        } finally {
            if (iterator != null) {
                try {
                    iterator.close();
                } catch (IOException e) {
                    throw new LocalStoreException(e);
                }
            }
            try {
                snapshot.close();
            } catch (IOException e) {
                throw new LocalStoreException(e);
            }
        }
    }

    @Override
    public void close() throws LocalStoreException {
        try {
            if (db != null) {
                db.close();
            }
        } catch (IOException e) {
            throw new LocalStoreException(e);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    public void destroy() throws LocalStoreException {
        try {
            close();
            JniDBFactory.factory.destroy(dbPath, options);
        } catch (IOException e) {
            throw new LocalStoreException(e);
        } finally {
            if (fileLock != null) {
                fileLock.release();
            }
            FileUtils.delete(dbPath);
        }
    }

}
